package com.yujiahao.bigdata.rdd.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  TODO combineByKey;这个第一个参数的数据类型是动态生成的，所以要将类型进行声明，也能实现WordCount（6 /10)
    >>rdd.combineByKey()参数说明 ：
      1)第一个参数表示相同K的第一个数据格式的转换；
      2)第二个参数：表示分区内计算规则；
      3)第三个参数：表示分区间的计算规则
 */

object Spark_RDD_Pper_Transform_CombineByKey {
  def main(args: Array[String]): Unit = {


  //TODO 1、获得连接
  val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
  val sc: SparkContext = new SparkContext(conf)


    val rdd= sc.makeRDD(List(("a",1),"a" -> 2,("b" -> 3),"b" -> 4,("b",5),"a" ->6),2)



   val value: RDD[(String, (Int, Int))] = rdd.combineByKey(
     (v) => (v, 1),
     (x: (Int, Int), v) => {
       (x._1 + v, x._2 + 1)
     },
     (t1: (Int, Int), t2: (Int, Int)) => {
       (t1._1 + t2._1, t1._2 + t2._2)
     }
   )
  //上面处理的结果是key,(v总和,Key出现的次数)
    value.mapValues(_._2).collect().foreach(println)
    //value.collect().foreach(println)
  //TODO 2、关闭连接
  sc.stop()
  }
}